package org.apache.rocketmq.client;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.rocketmq.common.message.Message;

public class ListSplitter implements Iterator<List<Message>> {

	private final int SIZE_LIMIT = 1000 * 1000;
	private final List<Message> messages;
	private int currIndex;

	public ListSplitter(List<Message> messages) {
		this.messages = messages;
	}

	@Override
	public boolean hasNext() {
		return currIndex < messages.size();
	}

	@Override
	public List<Message> next() {
		int nextIndex = currIndex;
		int totalSize = 0;
		for (; nextIndex < messages.size(); nextIndex++) {
			Message message = messages.get(nextIndex);
			int tmpSize = message.getTopic().length() + message.getBody().length;
			Map<String, String> properties = message.getProperties();
			for (Map.Entry<String, String> entry : properties.entrySet()) {
				tmpSize += entry.getKey().length() + entry.getValue().length();
			}
			tmpSize = tmpSize + 20; // for log overhead
			if (tmpSize > SIZE_LIMIT) {
				// it is unexpected that single message exceeds the SIZE_LIMIT
				// here just let it go, otherwise it will block the splitting
				// process
				if (nextIndex - currIndex == 0) {
					// if the next sublist has no element, add this one and then
					// break, otherwise just break
					nextIndex++;
				}
				break;
			}
			if (tmpSize + totalSize > SIZE_LIMIT) {
				break;
			} else {
				totalSize += tmpSize;
			}

		}
		List<Message> subList = messages.subList(currIndex, nextIndex);
		currIndex = nextIndex;
		return subList;
	}

	@Override
	public void remove() {

	}
}
